package com.example.monit.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.sql.QueryResponse;
import co.elastic.clients.json.JsonData;
import com.example.monit.bean.BasicIndicator;
import com.example.monit.bean.ResponseRust;
import com.example.monit.dto.QueryBasicIndicatorsDto;
import com.example.monit.dto.TotalDto;
import com.example.monit.enums.IndexName;
import com.example.monit.utils.TimeUtils;
import com.example.monit.vo.BasicindicatorsTotalVo;
import com.example.monit.vo.BasicindicatorsVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class BasicIndicatorServiceImpl extends BasEsService implements BasicIndicatorService {

    @Autowired
    ElasticsearchClient client;


    @Override
    public ResponseRust upLoadBasicIndicator(List<BasicIndicator> basicIndicator) throws IOException {
//        List<BulkOperation> operationList = basicIndicator.stream().map(e ->
//                        BulkOperation.of(builder ->
//                                builder.index(builder1 -> builder1.document(e).index(IndexName.BasicIndicator))))
//                .collect(Collectors.toList());
//
//        BulkResponse response = client.bulk(builder -> builder.operations(operationList));
//        if (response.errors()) {
//            System.out.println();
//            log.error("上报数据异常错误");
//        }
        return ResponseRust.success_creat();
    }


    @Override
    public ResponseRust queryBasicIndicator(BasicindicatorsVo basicindicatorsVo) throws IOException {
        String sql = " SELECT pageUrl,count(pageUrl),userID,sum(value) " +
                "            FROM \"basic_indicator\" " +
                "            where appId='%s' and mainType=%d and subType=%d and startTime between %d and %d" +
                "            group by pageUrl ,userID" +
                "            order by count(pageUrl) desc";
        sql = String.format(sql,
                basicindicatorsVo.getApp_id(),
                basicindicatorsVo.getMain_type(),
                basicindicatorsVo.getSub_type(),
                basicindicatorsVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                basicindicatorsVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli() //todo 时区问题
        );
        String finalSql = sql;
        QueryResponse query = client.sql().query(e -> e.query(finalSql).format("json"));
        List<List<JsonData>> list = query.rows();
        HashMap<String, QueryBasicIndicatorsDto> map = new HashMap<>();
        for (List<JsonData> jsonData : list) {
            String pageUrl = jsonData.get(0).to(String.class);
            Integer count = jsonData.get(1).to(Integer.class);
            String userID = jsonData.get(2).to(String.class);
            BigDecimal sumValue = jsonData.get(3).to(BigDecimal.class);

            LinkedList<String> userList = new LinkedList<>();
            LinkedList<String> pageList = new LinkedList<>();
            userList.add(userID);
            pageList.add(pageUrl);
            QueryBasicIndicatorsDto build = QueryBasicIndicatorsDto.builder()
                    .pageUrl(pageUrl)
                    .userList(userList)
                    .userCount(1)
                    .pageCount(1)
                    .pageList(pageList)
                    .sumAverage(sumValue)
                    .count(count).build();

            String key = pageUrl;
            QueryBasicIndicatorsDto queryDto = map.get(key);
            if (ObjectUtils.isEmpty(queryDto)) {
                map.put(pageUrl, build);
            } else {
                if (!queryDto.getUserList().contains(userID)) {
                    queryDto.setUserCount(queryDto.getUserCount() + 1);
                    queryDto.getUserList().add(userID);
                }
                if (!queryDto.getPageList().contains(pageUrl)) {
                    queryDto.setPageCount(queryDto.getPageCount() + 1);
                    queryDto.getPageList().add(pageUrl);
                }
                queryDto.setCount(queryDto.getCount() + build.getCount());
            }
        }

        Collection<QueryBasicIndicatorsDto> values = map.values();
        for (QueryBasicIndicatorsDto queryDto : values) {
            if (!queryDto.getSumAverage().equals(BigDecimal.ZERO)) {
                queryDto.setAverage(queryDto.getSumAverage().divide(new BigDecimal(queryDto.getCount()), 2, RoundingMode.HALF_UP));
            }
        }

        if (!ObjectUtils.isEmpty(basicindicatorsVo.getSize())) {
            values = values.stream().limit(basicindicatorsVo.getSize()).collect(Collectors.toList());
        }

        return ResponseRust.success(values);
    }

    @Override
    public ResponseRust totalBasicIndicator(BasicindicatorsTotalVo basicindicatorsTotalVo) throws IOException {
        SearchRequest searchRequest = SearchRequest.of(s ->
                s.index(IndexName.BasicIndicator)
                        .query(getQuery(basicindicatorsTotalVo))
                        .aggregations("count", getAggregation(basicindicatorsTotalVo))
                        .size(0));

        SearchResponse<BasicIndicator> response = client.search(searchRequest, BasicIndicator.class);

        Map<String, Aggregate> aggregations = response.aggregations();
        Aggregate count = aggregations.get("count");
        RangeAggregate range = count.range();

        LinkedList<List<TotalDto>> list = new LinkedList<>();
        // 5个范围
        for (RangeBucket bucket : range.buckets().array()) {
            Buckets<HistogramBucket> buckets = bucket.aggregations().get("list").histogram().buckets();
            list.add(totalData(buckets, basicindicatorsTotalVo));
        }
        return ResponseRust.success(list);
    }


    private Query getQuery(BasicindicatorsTotalVo basicindicatorsTotalVo) {
        return Query.of(query -> query.bool(
                bool -> bool
                        .must(t -> t.term(term -> term.field("appId").value(e -> e.stringValue(basicindicatorsTotalVo.getApp_id()))))
                        .must(t -> t.term(term -> term.field("mainType").value(e -> e.longValue(basicindicatorsTotalVo.getMain_type()))))
                        .must(t -> t.term(term -> term.field("subType").value(e -> e.longValue(basicindicatorsTotalVo.getSub_type()))))
                        .must(e -> e.range(r ->
                                r.gte(JsonData.of(basicindicatorsTotalVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                                        .lte(JsonData.of(basicindicatorsTotalVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                                        .field("startTime")))));
    }

    private Aggregation getAggregation(BasicindicatorsTotalVo basicindicatorsTotalVo) {
        List<AggregationRange> ranges = new LinkedList<>();
        ranges.add(AggregationRange.of(e -> e.from("0").to("1000")));
        ranges.add(AggregationRange.of(e -> e.from("1000").to("5000")));
        ranges.add(AggregationRange.of(e -> e.from("5000").to("10000")));
        ranges.add(AggregationRange.of(e -> e.from("10000").to("300000")));
        ranges.add(AggregationRange.of(e -> e.from("300000")));

        HashMap<String, Aggregation> aggregationHashMap = new HashMap<>(3);
        aggregationHashMap.put("avg", Aggregation.of(avg -> avg.avg(e -> e.field("value"))));
        aggregationHashMap.put("userCount", Aggregation.of(userCount -> userCount.cardinality(e -> e.field("userID"))));
        aggregationHashMap.put("pageCount", Aggregation.of(pageCount -> pageCount.cardinality(e -> e.field("pageUrl"))));

        if (!StringUtils.hasLength(basicindicatorsTotalVo.getGranularity())) {
            basicindicatorsTotalVo.setGranularity("1d");
        }
        int intValue = Integer.parseInt(basicindicatorsTotalVo.getGranularity().split("[smhdMy]")[0]);
        String type = basicindicatorsTotalVo.getGranularity().split(intValue + "")[1];
        ChronoUnit unit = TimeUtils.getType(type);

        return Aggregation.of(count -> count.range(RangeAggregation.of(e -> e.ranges(ranges).field("value")))
                .aggregations("list", list -> list.histogram(time -> time.interval((double) unit.getDuration().toMillis() * intValue)
                                .minDocCount(0)
                                .field("startTime"))
                        .aggregations(aggregationHashMap)));
    }


}
